package com.mng.processFun;

import com.mng.dto.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class TimerTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> stringDs = env.socketTextStream("localhost", 7777);
        SingleOutputStreamOperator<SensorReading> sensorDs = stringDs.map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String s) throws Exception {
                String[] strings = s.split(",");
                return new SensorReading(strings[0], Long.parseLong(strings[1]), Double.parseDouble(strings[2]));
            }
        });
        sensorDs.keyBy(sensorReading -> sensorReading.getId())
                .process(new MyProcessFun(Time.seconds(10).toMilliseconds()))
                .print();
        env.execute();

    }
}

class MyProcessFun extends KeyedProcessFunction<String, SensorReading , String>{

    //检查时间段
    private Long checkTime;
    //上一次温度
    ValueState<Double> lastTemp;
    //触发器的时间
    ValueState<Long>  recentTimer;
    public MyProcessFun(Long checkTime) {
        this.checkTime = checkTime;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        lastTemp = getRuntimeContext().getState(new ValueStateDescriptor<Double>("lastTem",Double.class));
        recentTimer = getRuntimeContext().getState(new ValueStateDescriptor<Long>("reTimer",Long.class));
    }

    @Override
    public void close() throws Exception {
        super.close();
        lastTemp.clear();
        recentTimer.clear();
    }

    @Override
    public void processElement(SensorReading sensorReading, Context context, Collector<String> collector) throws Exception {
        Double currTempV = sensorReading.getTemperaturre();
        Double lastTempV = lastTemp.value() != null? lastTemp.value() : currTempV;

        //如果温度升高且没有设置触发器
        //那么设置触发器并更新触发器时间状态
        if(currTempV > lastTempV && recentTimer.value() == null){
            context.timerService().registerProcessingTimeTimer(
                    context.timerService().currentProcessingTime()+checkTime);
            recentTimer.update(context.timerService().currentProcessingTime()+checkTime);

        }
        //如果温度不上升，那么删除触发器，并清空状态
        else if(currTempV <= lastTempV && recentTimer.value() != null){
            context.timerService().deleteProcessingTimeTimer(recentTimer.value());
            recentTimer.clear();
        }
        //更新上一次温度
        lastTemp.update(currTempV);

    }
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        // 触发报警，并且清除 定时器状态值
        out.collect("传感器" + ctx.getCurrentKey() + "温度值连续" + checkTime + "ms上升");
        recentTimer.clear();
    }
}
